package com.xian.java;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.MultipleParameterTool;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;

import java.util.Date;
import java.util.Iterator;

public class MyTest {
    public MyTest() {
    }

    public static void main(String[] args) throws Exception {
        MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);
        DataSet<String> text = null;
        if (params.has("input")) {
            Iterator var4 = params.getMultiParameterRequired("input").iterator();

            while(var4.hasNext()) {
                String input = (String)var4.next();
                if (text == null) {
                    text = env.readTextFile(input);
                } else {
                    text = ((DataSet)text).union(env.readTextFile(input));
                }
            }

            Preconditions.checkNotNull(text, "Input DataSet should not be null.");
        } else {
            System.out.println("Executing WordCount example with default input data set.");
            System.out.println("Use --input to specify file input.");

        }

        DataSet<Tuple2<String, Integer>> counts = ((DataSet)text).flatMap(new MyTest.Tokenizer()).groupBy(new int[]{0}).sum(1);
        if (params.has("output")) {
            counts.writeAsCsv(params.get("output"), "\n", " ");
            env.execute("WordCount Example");
        } else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            counts.print();
        }

    }

    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        public Tokenizer() {
        }

        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            //String[] tokens = value.toLowerCase().split("\\W+");
            String[] tokens = value.toLowerCase().split(" ");
            String[] var4 = tokens;
            int var5 = tokens.length;

            for(int var6 = 0; var6 < var5; ++var6) {
                String token = var4[var6];
                if (token.length() > 0) {
                    out.collect(new Tuple2(token, 1));
                }
            }

        }
    }
}
